package window;

import bean.SensorReading;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;


/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/4 10:11
 */
public class Window3_EventTimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * 从调用时刻开始给 env 创建的每一个 stream 追加时间特征
         */
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        /**
         * 每隔 0.1 秒产生一个 watermark
         */
        env.getConfig().setAutoWatermarkInterval(100);


        DataStream<String> inputStream = env.socketTextStream("192.168.88.106", 7777);

        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        })
                /**
                 * 升序数据设置时间和watermark
                 */
                /*.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() {
                    @Override
                    public long extractAscendingTimestamp(SensorReading sensorReading) {
                        return sensorReading.getTimestamp() * 1000L;
                    }
                })*/

                /**
                 * 乱序数据设置时间戳和watermark
                 */
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
                    @Override
                    public long extractTimestamp(SensorReading sensorReading) {
                        return sensorReading.getTimestamp() * 1000L;
                    }
                });


        /**
         * 数据偏移区间
         */
        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") {
        };

        /**
         * 基于事件时间的开窗聚合,统计15秒类温度最小值
         */

        SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id")
                .timeWindow(Time.seconds(15))
                .allowedLateness(Time.minutes(1))
                .sideOutputLateData(outputTag)
                .minBy("temperature");

        minTempStream.print("min");
        minTempStream.getSideOutput(outputTag).print("late");

        env.execute();

    }
}
